// Copyright(C) 2023 InfiniFlow, Inc. All rights reserved.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     https://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

module;

module infinity_core:expression_transformer.impl;

import :expression_transformer;

import :base_expression;
import :aggregate_expression;
import :between_expression;
import :cast_expression;
import :case_expression;
import :function_expression;
import :in_expression;
import :conjunction_expression;
import :expression_type;
import :infinity_exception;

import std;

namespace infinity {

std::vector<std::shared_ptr<BaseExpression>> SplitExpressionByDelimiter(const std::shared_ptr<BaseExpression> &expression,
                                                                        ConjunctionType delimiter) {
    std::vector<std::shared_ptr<BaseExpression>> result;

    std::function<VisitControlType(std::shared_ptr<BaseExpression> & child)> func =
        [&](const std::shared_ptr<BaseExpression> &expr_ptr) -> VisitControlType {
        if (expr_ptr->type() == ExpressionType::kConjunction) {
            const auto conjunction_expr_ptr = std::static_pointer_cast<ConjunctionExpression>(expr_ptr);
            if (conjunction_expr_ptr->conjunction_type() == delimiter)
                return VisitControlType::kVisit;
        }
        result.emplace_back(expr_ptr);
        return VisitControlType::kNotVisit;
    };

    VisitExpression(expression, func);

    return result;
}

std::shared_ptr<BaseExpression> ComposeExpressionWithDelimiter(const std::vector<std::shared_ptr<BaseExpression>> &expressions,
                                                               ConjunctionType conjunction_type) {
    auto expr_count = expressions.size();
    if (expr_count == 0) {
        return nullptr;
    }
    std::shared_ptr<BaseExpression> result = expressions[0];
    for (size_t i = 1; i < expr_count; ++i) {
        result = std::make_shared<ConjunctionExpression>(conjunction_type, result, expressions[i]);
    }
    return result;
}

void VisitExpression(const std::shared_ptr<BaseExpression> &expression,
                     const std::function<VisitControlType(std::shared_ptr<BaseExpression> &child)> &visitor) {
    std::queue<std::shared_ptr<BaseExpression>> queue;
    queue.push(expression);

    while (!queue.empty()) {
        auto expr = queue.front();
        queue.pop();

        if (visitor(expr) == VisitControlType::kVisit) {
            for (auto &argument : expr->arguments()) {
                queue.push(argument);
            }
        }
    }
}

void VisitExpression(const std::shared_ptr<BaseExpression> &expression, const std::function<void(std::shared_ptr<BaseExpression> &child)> &visitor) {
    switch (expression->type()) {
        case ExpressionType::kAggregate: {
            auto *agg_expr = static_cast<AggregateExpression *>(expression.get());
            for (auto &argument : agg_expr->arguments()) {
                visitor(argument);
            }
            break;
        }
        case ExpressionType::kBetween: {
            auto *between_expr = static_cast<BetweenExpression *>(expression.get());
            for (auto &argument : between_expr->arguments()) {
                visitor(argument);
            }
            break;
        }
        case ExpressionType::kCast: {
            auto *cast_expr = static_cast<CastExpression *>(expression.get());
            for (auto &argument : cast_expr->arguments()) {
                visitor(argument);
            }
            break;
        }
        case ExpressionType::kCase: {
            auto *case_expr = static_cast<CaseExpression *>(expression.get());
            for (auto &argument : case_expr->arguments()) {
                visitor(argument);
            }
            for (auto &case_when : case_expr->CaseExpr()) {
                visitor(case_when.when_expr_);
                visitor(case_when.then_expr_);
            }
            visitor(case_expr->ElseExpr());
            break;
        }

        case ExpressionType::kConjunction: {
            auto *conjunction_expr = static_cast<ConjunctionExpression *>(expression.get());
            for (auto &argument : conjunction_expr->arguments()) {
                visitor(argument);
            }
            break;
        }
        case ExpressionType::kFunction: {
            auto *function_expr = static_cast<FunctionExpression *>(expression.get());
            for (auto &argument : function_expr->arguments()) {
                visitor(argument);
            }
            break;
        }
        case ExpressionType::kIn: {
            auto *function_expr = static_cast<InExpression *>(expression.get());
            visitor(function_expr->left_operand());

            for (auto &argument : function_expr->arguments()) {
                visitor(argument);
            }
            break;
        }
        case ExpressionType::kSubQuery:
            break;
        case ExpressionType::kColumn:
        case ExpressionType::kValue:
        case ExpressionType::kFilterFullText:
            break;
        default: {
            UnrecoverableError(fmt::format("Unsupported expression type: {}", expression->Name()));
        }
    }
}

} // namespace infinity